/* header file */
#include "threadpool_02.h"

/* variable definition  */
/* function declaration */
int tpool_init(threadpool_t* tpool, size_t start_threadnum); 
int tpool_destroy(threadpool_t* tpool);
static void** tpool_join(threadpool_t* tpool, pthread_t sleftid);
static int submit_task(threadpool_t* tpool, thread_task_t* task);
static void* worker(void*);
static void* thread_exit(void*);



int tpool_init(threadpool_t* tpool, size_t start_threadnum)
{
    // 初始化线程池属性
    memset(tpool, 0, sizeof(threadpool_t));
   
    // 任务缓冲队列
    bufque_init(&tpool->mission_que, MAXTASK, sizeof(thread_task_t));
    
    // 为线程信息表分配内存
    tpool->threads = (thread_t*)malloc(start_threadnum * sizeof(thread_t));
    assert(tpool->threads != NULL);

    // 设置线程池属性
    tpool->capacity = start_threadnum;
    tpool->size = start_threadnum;
    pthread_mutex_init(&tpool->threads_lock, NULL);
    tpool->submit_task = submit_task;
    
    // 启动所有线程（可以考虑将启动线程数作为参数）
    for(size_t i=0; i<start_threadnum; i++){
        pthread_create( &tpool->threads[i].tid, NULL, 
                        worker, (void*)&tpool->mission_que);
        printf("thread %lu boot\n", tpool->threads[i].tid);
    }
    
    // 报告进程池启动成功（可转为日志）
    fprintf(stdout, "The %ld threads pool has been started.\n\n", start_threadnum);
    
    return 0;
}

static void** tpool_join(threadpool_t* tpool, pthread_t selftid)
{// selftid 是发起join的线程自身tid
    void** rets =   (void**)calloc
                    (tpool->capacity, sizeof(void*));   // capacity or size ?
    for(size_t i=0;i<tpool->capacity;i++){
        if(tpool->threads[i].tid == selftid){  continue;}  // 不要等待自身退出陷入死循环
        pthread_join(tpool->threads[i].tid, &rets[i]);
    }
    
    return rets;
}

static int submit_task(threadpool_t* tpool, thread_task_t* task)
{
    sem_wait(&tpool->mission_que.empty);
    sem_wait(&tpool->mission_que.lock);
    
    assert(tpool->mission_que.size < MAXTASK);
    tpool->mission_que.enque(&tpool->mission_que, (void*)task);
    
    sem_post(&tpool->mission_que.lock);
    sem_post(&tpool->mission_que.full);

    puts("threadpool.c\\submit_task:"
         "mission submitted");
    return 0;
}

static void* worker(void* args)
{
    //puts("fafafa, famugong");
    thread_task_t mymission;
    bufque_t* miss = (bufque_t*)args;
     
    while(1){
        sem_wait(&miss->full);
        sem_wait(&miss->lock);

        void* temp = miss->deque(miss);
        mymission = *((thread_task_t*)temp);

        sem_post(&miss->lock);
        sem_post(&miss->empty);

        printf("threadpool.c\\worker(): "
               "thread %lu get routine %p with args %p\n", 
               pthread_self(), mymission.routine, mymission.args);
        mymission.routine(mymission.args);
        
        // clean up
        free(temp); // 注意，temp接收的值在缓冲队列中被定义为指向堆上内存的指针，要求在外部做内存管理
        
        printf("thread %lu finished routine %p with args %p\n", 
               pthread_self(), mymission.routine, mymission.args);
    }   
   
    return NULL;
};

static void* thread_exit(void* args)
{
    threadpool_t* tpool = (threadpool_t*)args;
    
    pthread_mutex_lock(&tpool->threads_lock);
    tpool->size--;
    pthread_mutex_unlock(&tpool->threads_lock);
    
    printf("thread %lu exit\n", pthread_self());
    sleep(1);
    pthread_exit(NULL);
}


int tpool_destroy(threadpool_t* tpool)
{ 
    // 向任务队列中提交足够多的退出任务
    // 注意，如发起队列销毁的线程自身就是线程池中的工作线程
    // 自身接取退出任务将会引发严重bug
    thread_task_t task;
    task.routine = thread_exit;
    task.args = NULL;
    for(size_t i=0; i<tpool->capacity; i++){
        tpool->submit_task(tpool, &task);
    }
    
    // 等待所有接取任务的工作线程主动退出（可以加入强制退出指令）
    tpool_join(tpool, pthread_self());
    
    // 所有工作进程退出后开始销毁进程池，释放内存空间
    bufque_destroy(&tpool->mission_que);
    free(tpool->threads);
    pthread_mutex_destroy(&tpool->threads_lock);
    
    return 0;
};
